package data.platform.store.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.rocksdb.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;

@Configuration
@ConditionalOnProperty(prefix = "checkpoint", name = "store", havingValue = "rocksdb")
@Slf4j
public class RocksDBConfig {

    @Value("${rocksdb.save_dir}")
    private String saveDir;

    @Value("${rocksdb.disableWAL}")
    private boolean disableWAL;

    private static ObjectMapper objectMapper = new ObjectMapper();

    private static final CompressionType COMPRESSION_TYPE = CompressionType.NO_COMPRESSION;
    private static final CompactionStyle COMPACTION_STYLE = CompactionStyle.UNIVERSAL;
    private static final long WRITE_BUFFER_SIZE = 32 * 1024 * 1024L;
    private static final int MAX_WRITE_BUFFERS = 3;
    private static final String DB_FILE_DIR = "rocksdb";

    private WriteOptions wOptions;

    private FlushOptions fOptions;

    private RocksDB db;

    @PostConstruct
    public void init() {
        log.info("enable rocksdb.........");
        RocksDB.loadLibrary();
    }

    @Bean
    public WriteOptions writeOptions() {
        wOptions  = new WriteOptions();
        wOptions.setDisableWAL(disableWAL);
        return wOptions;
    }

    @Bean
    public FlushOptions flushOptions() {
        fOptions = new FlushOptions();
        fOptions.setWaitForFlush(true);
        return fOptions;
    }

    @Bean
    public RocksDB rocksDB() throws Exception {

        Options options = new Options();
        options.setWriteBufferSize(WRITE_BUFFER_SIZE);
        options.setCompressionType(COMPRESSION_TYPE);
        options.setCompactionStyle(COMPACTION_STYLE);
        options.setMaxWriteBufferNumber(MAX_WRITE_BUFFERS);
        options.setCreateIfMissing(true);
        options.setErrorIfExists(false);

        try {
            Path saveDirPath = Paths.get(saveDir);
            if (Files.notExists(saveDirPath)) {
                Files.createDirectories(saveDirPath);
            }

            Path dbPath = saveDirPath.resolve(DB_FILE_DIR);
            if (Files.notExists(dbPath)) {
                Files.createDirectories(dbPath);
            }
            db = RocksDB.open(options, dbPath.toFile().getAbsolutePath());
            return db;
        } catch (IOException | RocksDBException ex) {
            log.error("initializing RocksDB error",ex);
            throw new RuntimeException("initializing RocksDB error");
        }
    }

    @PreDestroy
    public void onDestroy() throws Exception {
        log.info("close rocksdb......");
        wOptions.close();
        fOptions.close();
        db.close();
    }

}
